package cn.edu.nchu.deviceSimulation;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class Device implements Runnable{
    private static String broker = "ws://175.24.184.18:8083/mqtt";
    private static final String[] SUSCRIBE_TOPICS = {"messages/down", "commands/request"};
    private final String PRODUCT_ID;
    private final String DEVICE_ID;
    private final String DEVICE_PASSWORD;
    private final String TOPIC_PREFIX;
    private MqttClient client;

    public Device(String product_id, String device_id, String device_password) {
        PRODUCT_ID = product_id;
        DEVICE_ID = device_id;
        DEVICE_PASSWORD = device_password;
        TOPIC_PREFIX = "/" + PRODUCT_ID + "/" + DEVICE_ID + "/";
    }


    @Override
    public void run() {
        try {
            System.out.println(TOPIC_PREFIX + "------启动了！！！");
            this.init();
            cycleSendMessage();
        } catch (MqttException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void init() throws MqttException {
        initClient();
        connectMqtt();
        // 订阅
        for (String topic : SUSCRIBE_TOPICS) {
            client.subscribe(TOPIC_PREFIX + topic);
        }
    }

    private void cycleSendMessage() throws InterruptedException {
        while (true) {
            Thread.sleep(5*1000);
            String mqtt_message = "{ \"msg\": \""+ "Current Time:" + System.currentTimeMillis() + "\"}";
            sendMessage(mqtt_message, TOPIC_PREFIX + "messages/up");
        }
    }

    public void sendMessage(String content, String topic){
        MqttMessage message = new MqttMessage(content.getBytes());
//        message.setRetained(true);
        try {
            client.publish(topic, message);
        }catch (MqttException ignored){

        }
    }

    private void connectMqtt() throws MqttException {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        connectOptions.setUserName(DEVICE_ID);
        connectOptions.setPassword(DEVICE_PASSWORD.toCharArray());
        // 保留会话
//        connectOptions.setCleanSession(true);
        // 建立连接
        client.connect(connectOptions);
    }

    private void initClient() throws MqttException {
        client = new MqttClient(broker, DEVICE_ID, new MemoryPersistence());
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                // 连接丢失后，一般在这里面进行重连
                try {
                    client.reconnect();
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                // subscribe后得到的消息会执行到这里面
                messageHandler(topic, message);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                // 发布消息后要做的事
            }
        });
    }

    private void messageHandler(String topic, MqttMessage message){
        System.out.println(topic);
        System.out.println(message);
    }
}
